#define _GNU_SOURCE         // 添加到文件最开始
#define _POSIX_C_SOURCE 200809L
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <signal.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include<strings.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>
#include <raft.h>
#include <raft/uv.h>

#define MAX_EVENTS 10
#define BUFFER_SIZE 1024


// 添加日志相关定义
#define LOG_DEBUG 0
#define LOG_INFO 1
#define LOG_WARN 2
#define LOG_ERROR 3

static FILE *log_fp = NULL;

// 日志初始化函数
void init_logger(const char *log_file) {
  log_fp = fopen(log_file, "a");
  if (log_fp == NULL) {
    fprintf(stderr, "Failed to open log file: %s\n", log_file);
    exit(1);
  }
  setvbuf(log_fp, NULL, _IOLBF, 0);  // 行缓冲
}

// 日志函数
void server_log(int level, const char *fmt, ...) {
  if (!log_fp) return;

  time_t now;
  time(&now);
  struct tm *tm_info = localtime(&now);
  char timestamp[26];
  strftime(timestamp, sizeof(timestamp), "%Y-%m-%d %H:%M:%S", tm_info);

  const char *level_str;
  switch (level) {
    case LOG_DEBUG:
      level_str = "DEBUG";
      break;
    case LOG_INFO:
      level_str = "INFO";
      break;
    case LOG_WARN:
      level_str = "WARN";
      break;
    case LOG_ERROR:
      level_str = "ERROR";
      break;
    default:
      level_str = "UNKNOWN";
      break;
  }

  fprintf(log_fp, "[%s] [%s] ", timestamp, level_str);

  va_list args;
  va_start(args, fmt);
  vfprintf(log_fp, fmt, args);
  va_end(args);
  fprintf(log_fp, "\n");
  fflush(log_fp);
}

// 全局变量用于控制程序退出
volatile sig_atomic_t running = 1;
int server_fd = -1;
int epoll_fd = -1;

#define MAX_ENTRY_STR_SIZE 128
#define MAX_MAP_SIZE 1000
typedef struct redis_map_entry {
  char key[MAX_ENTRY_STR_SIZE];
  char value[MAX_ENTRY_STR_SIZE];
} redis_map_entry;
typedef struct redis_map_t {
  redis_map_entry *map[MAX_MAP_SIZE];
  size_t size;
} redis_map_t;
redis_map_t redis_map = {.map = {0}, .size = 0};

// 在全局变量区域添加配置
typedef struct {
  char dir[256];
  char dbfilename[256];
  int id;
  int m;
} server_config;

server_config config = {
    .dir = "./",              // 默认值
    .dbfilename = "dump.rdb",  // 默认值
    .id = 0,                   //server id
    .m = 6389                  //redis main port
};


#define N_SERVERS 3    /* Number of servers in the example cluster */
#define APPLY_RATE 125 /* Apply a new entry every 125 milliseconds */

struct Fsm {
  char key[MAX_ENTRY_STR_SIZE];
  char value[MAX_ENTRY_STR_SIZE];
};

static int FsmApply(struct raft_fsm *fsm,
                    const struct raft_buffer *buf,
                    void **result)
{
    struct Fsm *f = fsm->data;
    // 按照rdb格式中的stringencoding将buf中的数据复制到f->key和f->value，前4个字节是长度，后4个字节是数据
    uint32_t key_len = *(uint32_t *)buf->base;
    memcpy(f->key, buf->base+4, key_len);
    uint32_t value_len = *(uint32_t *)(buf->base + 4 + key_len);
    memcpy(f->value, buf->base + key_len + 8, value_len);
    return 0;
}

static int FsmSnapshot(struct raft_fsm *fsm,
                       struct raft_buffer *bufs[],
                       unsigned *n_bufs)
{
    struct Fsm *f = fsm->data;
    *n_bufs = 1;
    *bufs = raft_malloc(sizeof **bufs);
    if (*bufs == NULL) {
        return RAFT_NOMEM;
    }
    (*bufs)[0].len = sizeof(uint64_t);
    (*bufs)[0].base = raft_malloc((*bufs)[0].len);
    if ((*bufs)[0].base == NULL) {
        return RAFT_NOMEM;
    }
    uint32_t key_len = strlen(f->key);
    uint32_t value_len = strlen(f->value);
    uint32_t total_len = key_len + value_len + 8;
    memcpy((*bufs)[0].base, &total_len, 4);
    memcpy((*bufs)[0].base + 4, f->key, key_len);
    memcpy((*bufs)[0].base + 4 + key_len, &value_len, 4);
    memcpy((*bufs)[0].base + 4 + key_len + 4, f->value, value_len);
    printf("FsmSnapshot: key=%s, value=%s\n", f->key, f->value);
    return 0;
}

static int FsmRestore(struct raft_fsm *fsm, struct raft_buffer *buf)
{
    struct Fsm *f = fsm->data;
    if (buf->len != sizeof(uint64_t)) {
        return RAFT_MALFORMED;
    }
    uint32_t key_len = *(uint32_t *)buf->base;
    memcpy(f->key, buf->base+4, key_len);
    uint32_t value_len = *(uint32_t *)(buf->base + 4 + key_len);
    memcpy(f->value, buf->base + key_len + 8, value_len);
    raft_free(buf->base);
    return 0;
}

static int FsmInit(struct raft_fsm *fsm)
{
    struct Fsm *f = raft_malloc(sizeof *f);
    if (f == NULL) {
        return RAFT_NOMEM;
    }
    memset(f->key, 0, sizeof(f->key));
    memset(f->value, 0, sizeof(f->value));
    fsm->version = 2;
    fsm->data = f;
    fsm->apply = FsmApply;
    fsm->snapshot = FsmSnapshot;
    fsm->snapshot_finalize = NULL;
    fsm->restore = FsmRestore;
    return 0;
}

static void FsmClose(struct raft_fsm *f)
{
    if (f->data != NULL) {
        raft_free(f->data);
    }
}

////////////////////////////////////////////////////////////////////////

struct Server;
typedef void (*ServerCloseCb)(struct Server *server);

struct Server
{
    void *data;                         /* User data context. */
    struct uv_loop_s *loop;             /* UV loop. */
    struct uv_timer_s timer;            /* To periodically apply a new entry. */
    const char *dir;                    /* Data dir of UV I/O backend. */
    struct raft_uv_transport transport; /* UV I/O backend transport. */
    struct raft_io io;                  /* UV I/O backend. */
    struct raft_fsm fsm;                /* Sample application FSM. */
    unsigned id;                        /* Raft instance ID. */
    char address[64];                   /* Raft instance address. */
    struct raft raft;                   /* Raft instance. */
    struct raft_transfer transfer;      /* Transfer leadership request. */
    ServerCloseCb close_cb;             /* Optional close callback. */
};

static void serverRaftCloseCb(struct raft *raft)
{
    struct Server *s = raft->data;
    raft_uv_close(&s->io);
    raft_uv_tcp_close(&s->transport);
    FsmClose(&s->fsm);
    if (s->close_cb != NULL) {
        s->close_cb(s);
    }
}

static void serverTransferCb(struct raft_transfer *req)
{
    struct Server *s = req->data;
    raft_id id;
    const char *address;
    raft_leader(&s->raft, &id, &address);
    raft_close(&s->raft, serverRaftCloseCb);
}

/* Final callback in the shutdown sequence, invoked after the timer handle has
 * been closed. */
static void serverTimerCloseCb(struct uv_handle_s *handle)
{
    struct Server *s = handle->data;
    if (s->raft.data != NULL) {
        if (s->raft.state == RAFT_LEADER) {
            int rv;
            rv = raft_transfer(&s->raft, &s->transfer, 0, serverTransferCb);
            if (rv == 0) {
                return;
            }
        }
        raft_close(&s->raft, serverRaftCloseCb);
    }
}

/* Initialize the example server struct, without starting it yet. */
static int ServerInit(struct Server *s,
                      struct uv_loop_s *loop,
                      const char *dir,
                      unsigned id)
{
    struct raft_configuration configuration;
    struct timespec now;
    unsigned i;
    int rv;

    memset(s, 0, sizeof *s);

    /* Seed the random generator */
    timespec_get(&now, TIME_UTC);
    srand((unsigned)(now.tv_nsec ^ now.tv_sec));

    s->loop = loop;

    /* Add a timer to periodically try to propose a new entry. */
    rv = uv_timer_init(s->loop, &s->timer);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: uv_timer_init(): %s", id, uv_strerror(rv));
        goto err;
    }
    s->timer.data = s;

    /* Initialize the TCP-based RPC transport. */
    s->transport.version = 1;
    s->transport.data = NULL;
    rv = raft_uv_tcp_init(&s->transport, s->loop);
    if (rv != 0) {
        goto err;
    }

    /* Initialize the libuv-based I/O backend. */
    rv = raft_uv_init(&s->io, s->loop, dir, &s->transport);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: raft_uv_init(): %s", id, s->io.errmsg);
        goto err_after_uv_tcp_init;
    }

    /* Initialize the finite state machine. */
    rv = FsmInit(&s->fsm);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: FsmInit(): %s", id, raft_strerror(rv));
        goto err_after_uv_init;
    }

    /* Save the server ID. */
    s->id = id;

    /* Render the address. */
    sprintf(s->address, "127.0.0.1:900%d", id);

    /* Initialize and start the engine, using the libuv-based I/O backend. */
    rv = raft_init(&s->raft, &s->io, &s->fsm, id, s->address);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: raft_init(): %s", id, raft_errmsg(&s->raft));
        goto err_after_fsm_init;
    }
    s->raft.data = s;

    /* Bootstrap the initial configuration if needed. */
    raft_configuration_init(&configuration);
    for (i = 0; i < N_SERVERS; i++) {
        char address[64];
        unsigned server_id = i + 1;
        sprintf(address, "127.0.0.1:900%d", server_id);
        rv = raft_configuration_add(&configuration, server_id, address,
                                    RAFT_VOTER);
        if (rv != 0) {
            server_log(LOG_INFO, "%d: raft_configuration_add(): %s", id, raft_strerror(rv));
            goto err_after_configuration_init;
        }
    }
    rv = raft_bootstrap(&s->raft, &configuration);
    if (rv != 0 && rv != RAFT_CANTBOOTSTRAP) {
        goto err_after_configuration_init;
    }
    raft_configuration_close(&configuration);

    raft_set_snapshot_threshold(&s->raft, 64);
    raft_set_snapshot_trailing(&s->raft, 16);
    raft_set_pre_vote(&s->raft, true);

    s->transfer.data = s;

    return 0;

err_after_configuration_init:
    raft_configuration_close(&configuration);
err_after_fsm_init:
    FsmClose(&s->fsm);
err_after_uv_init:
    raft_uv_close(&s->io);
err_after_uv_tcp_init:
    raft_uv_tcp_close(&s->transport);
err:
    return rv;
}

/* Called after a request to apply a new command to the FSM has been
 * completed. */
static void serverApplyCb(struct raft_apply *req, int status, void *result)
{
    struct Server *s = req->data;
    int count;
    raft_free(req);
    if (status != 0) {
        if (status != RAFT_LEADERSHIPLOST) {
            server_log(LOG_INFO, "%d: raft_apply() callback: %s (%d)", s->id, raft_errmsg(&s->raft),
                 status);
        }
        return;
    }
    count = *(int *)result;
    if (count % 100 == 0) {
        server_log(LOG_INFO, "%d: count %d", s->id, count);
    }
}

/* Called periodically every APPLY_RATE milliseconds. */
static void serverTimerCb(uv_timer_t *timer)
{
    struct Server *s = timer->data;
    struct raft_buffer buf;
    struct raft_apply *req;
    int rv;

    if (s->raft.state != RAFT_LEADER) {
        return;
    }

    buf.len = sizeof(uint64_t);
    buf.base = raft_malloc(buf.len);
    if (buf.base == NULL) {
        server_log(LOG_INFO, "%d: serverTimerCb(): out of memory", s->id);
        return;
    }

    *(uint64_t *)buf.base = 1;

    req = raft_malloc(sizeof *req);
    if (req == NULL) {
        server_log(LOG_INFO, "%d: serverTimerCb(): out of memory", s->id);
        return;
    }
    req->data = s;

    rv = raft_apply(&s->raft, req, &buf, 1, serverApplyCb);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: raft_apply(): %s", s->id, raft_errmsg(&s->raft));
        return;
    }
}

/* Start the example server. */
static int ServerStart(struct Server *s)
{
    int rv;

    server_log(LOG_INFO, "%d: starting", s->id);

    rv = raft_start(&s->raft);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: raft_start(): %s", s->id, raft_errmsg(&s->raft));
        goto err;
    }
    rv = uv_timer_start(&s->timer, serverTimerCb, 0, 125);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: uv_timer_start(): %s", s->id, uv_strerror(rv));
        goto err;
    }

    return 0;

err:
    return rv;
}

/* Release all resources used by the example server. */
static void ServerClose(struct Server *s, ServerCloseCb cb)
{
    s->close_cb = cb;

    server_log(LOG_INFO, "%d: stopping", s->id);

    /* Close the timer asynchronously if it was successfully
     * initialized. Otherwise invoke the callback immediately. */
    if (s->timer.data != NULL) {
        uv_close((struct uv_handle_s *)&s->timer, serverTimerCloseCb);
    } else {
        s->close_cb(s);
    }
}

/********************************************************************
 *
 * Top-level main loop.
 *
 ********************************************************************/

static void mainServerCloseCb(struct Server *server)
{
    struct uv_signal_s *sigint = server->data;
    uv_close((struct uv_handle_s *)sigint, NULL);
}

/* Handler triggered by SIGINT. It will initiate the shutdown sequence. */
static void mainSigintCb(struct uv_signal_s *handle, int signum)
{
    (void)signum;
    struct Server *server = handle->data;
    assert(signum == SIGINT);
    uv_signal_stop(handle);
    server->data = handle;
    ServerClose(server, mainServerCloseCb);
}

int init_raft(int id, char* dir)
{
    struct uv_loop_s loop;
    struct uv_signal_s sigint; /* To catch SIGINT and exit. */
    struct Server server;
    int rv;

    /* Ignore SIGPIPE, see https://github.com/joyent/libuv/issues/1254 */
    signal(SIGPIPE, SIG_IGN);

    /* Initialize the libuv loop. */
    rv = uv_loop_init(&loop);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: uv_loop_init(): %s", id, uv_strerror(rv));
        goto err;
    }

    /* Initialize the example server. */
    rv = ServerInit(&server, &loop, dir, id);
    if (rv != 0) {
        goto err_after_server_init;
    }

    /* Add a signal handler to stop the example server upon SIGINT. */
    rv = uv_signal_init(&loop, &sigint);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: uv_signal_init(): %s", id, uv_strerror(rv));
        goto err_after_server_init;
    }
    sigint.data = &server;
    rv = uv_signal_start(&sigint, mainSigintCb, SIGINT);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: uv_signal_start(): %s", id, uv_strerror(rv));
        goto err_after_signal_init;
    }

    /* Start the server. */
    rv = ServerStart(&server);
    if (rv != 0) {
        goto err_after_signal_init;
    }

    /* Run the event loop until we receive SIGINT. */
    rv = uv_run(&loop, UV_RUN_DEFAULT);
    if (rv != 0) {
        server_log(LOG_INFO, "%d: uv_run_start(): %s", id, uv_strerror(rv));
    }

    uv_loop_close(&loop);

    return rv;

err_after_signal_init:
    uv_close((struct uv_handle_s *)&sigint, NULL);
err_after_server_init:
    ServerClose(&server, NULL);
    uv_run(&loop, UV_RUN_DEFAULT);
    uv_loop_close(&loop);
err:
    return rv;
}

////////////////////////////////////////////////////////////////////////

// 信号处理函数
void handle_signal(int signum) {
  if (signum == SIGTERM || signum == SIGINT) {
    running = 0;
    // 关闭 server socket，这会导致 accept 循环退出
    if (server_fd != -1) {
      close(server_fd);
    }
    server_log(LOG_INFO, "Received signal %d, shutting down...", signum);
  }
}

void daemonize() {
  pid_t pid = fork();
  if (pid < 0) {
    exit(EXIT_FAILURE);
  }
  if (pid > 0) {
    exit(EXIT_SUCCESS);
  }

  if (setsid() < 0) {
    exit(EXIT_FAILURE);
  }

  signal(SIGCHLD, SIG_IGN);

  pid = fork();
  if (pid < 0) {
    exit(EXIT_FAILURE);
  }
  if (pid > 0) {
    exit(EXIT_SUCCESS);
  }

  umask(0);
  chdir("/");

  for (int x = sysconf(_SC_OPEN_MAX); x >= 0; x--) {
    close(x);
  }
}

// 设置非阻塞模式
void set_nonblocking(int sock) {
  int flags = fcntl(sock, F_GETFL, 0);
  fcntl(sock, F_SETFL, flags | O_NONBLOCK);
}

void send_msg(int sd, const char *msg) {
  char msg_formatted[256];
  msg_formatted[0] = '\0';
  snprintf(msg_formatted, sizeof(msg_formatted), "$%zu\r\n%s\r\n", strlen(msg),
           msg);
  send(sd, msg_formatted, strlen(msg_formatted), 0);
  server_log(LOG_INFO, "Send msg: %s", msg_formatted);
}

// RDB 文件格式常量
#define RDB_MAGIC "REDIS"
#define RDB_VERSION "0011"
#define RDB_TYPE_STRING 0
#define RDB_OPCODE_EOF 255
#define RDB_OPCODE_SELECTDB 254  // 选择数据库的操作码
#define RDB_OPCODE_RESIZEDB 251  // 数据库大小信息的操作码
#define RDB_MAGIC_STRING "REDIS"

// RDB 文件头部结构
struct rdb_header {
  char magic[5];    // "REDIS"
  char version[4];  // "0011"
};

// 从文件读取指定长度的数据
size_t rdb_read(FILE *fp, void *buf, size_t len) {
  return fread(buf, 1, len, fp);
}

// 添加字节序处理函数
uint32_t load_uint32(FILE *fp) {
  uint32_t value;
  uint8_t buf[4];

  if (rdb_read(fp, buf, 4) != 4) {
    return 0;
  }

  // 从小端序字节构建 uint32
  value = ((uint32_t)buf[0]) | ((uint32_t)buf[1] << 8) |
          ((uint32_t)buf[2] << 16) | ((uint32_t)buf[3] << 24);

  return value;
}

// 加载 RDB 文件
int load_rdb_file() {
  char filepath[512];
  snprintf(filepath, sizeof(filepath), "%s/%s", config.dir, config.dbfilename);

  FILE *fp = fopen(filepath, "rb");
  if (!fp) {
    server_log(LOG_INFO, "No RDB file found at %s", filepath);
    return 0;
  }

  // 读取并验证文件头
  struct rdb_header header;
  char magic[6] = {0};
  char version[5] = {0};
  fread(magic, 5, 1, fp);
  fread(version, 4, 1, fp);

  // 验证 magic string
  if (memcmp(magic, RDB_MAGIC_STRING, 5) != 0) {
    server_log(LOG_ERROR, "Invalid RDB file format");
    fclose(fp);
    return -1;
  }

  // 验证版本
  int iversion;
  sscanf(version, "%d", &iversion);
  if (iversion != 11) {
    server_log(LOG_ERROR, "Unsupported RDB version: %d", version);
    fclose(fp);
    return -1;
  }

  while (!feof(fp)) {
    uint8_t type;
    if (rdb_read(fp, &type, 1) != 1) {
      break;
    }

    if (type == RDB_OPCODE_EOF) {
      break;
    } else if (type == RDB_OPCODE_SELECTDB) {
      uint8_t dbid;
      if (rdb_read(fp, &dbid, 1) != 1) {
        server_log(LOG_ERROR, "Failed to read database number");
        fclose(fp);
        return -1;
      }
      server_log(LOG_DEBUG, "Selecting database: %d", dbid);
    } else {
      fseek(fp, -1, SEEK_CUR);
      break;
    }
  }

  // 读取键值对
  while (!feof(fp)) {
    uint8_t type;
    if (rdb_read(fp, &type, 1) != 1) {
      break;
    }

    if (type == RDB_OPCODE_EOF) {
      break;
    }else if(type == 0xFB){
      uint8_t entrysize = 0;
      rdb_read(fp, &entrysize, 1);
      server_log(LOG_DEBUG, "Entry size: %d", entrysize);
      for(int i = 0; i < entrysize; i++){
        uint8_t valuetye = 0;
        rdb_read(fp, &valuetye, 1);
        server_log(LOG_DEBUG, "Value type: %d", valuetye);
        uint32_t keylen = 0;
        rdb_read(fp, &keylen, 4);
        char key[keylen];
        rdb_read(fp, key, keylen);
        server_log(LOG_DEBUG, "Key: %s", key);
        uint32_t valuelen = 0;
        rdb_read(fp, &valuelen, 4);
        char value[valuelen];
        rdb_read(fp, value, valuelen);
        server_log(LOG_DEBUG, "Value: %s", value);
        redis_map_entry *entry =
            (redis_map_entry *)malloc(sizeof(redis_map_entry));
        strcpy(entry->key, key);
        strcpy(entry->value, value);
        redis_map.map[redis_map.size++] = entry;
      }
    }
  }

  server_log(LOG_INFO, "Loaded %zu entries from RDB file", redis_map.size);
  fclose(fp);
  return 0;
}

// 写入 RDB 文件头
void write_rdb_header(FILE *file) {
    char magic[6] = {0}; // magic 字符串以空字符结尾，所以需要6个字节的空间
    memcpy(magic, RDB_MAGIC, 5); // 复制 magic 字符串，不包括空字符
    fwrite(magic, 1, 5, file); // 写入 magic 字符串
    char version[5] = {0};
    memcpy(version, RDB_VERSION, 4);
    fwrite(version, 1, 4, file); // 写入版本号
    uint8_t dbindex = 0xFE;
    fwrite(&dbindex, sizeof(uint8_t), 1, file);
    uint8_t index = 0; // 假设只有一个数据库
    fwrite(&index, sizeof(uint32_t), 1, file);
}

// 写入字符串对象到 RDB 文件
void write_string_object(FILE *file, const char *str, size_t len) {
  uint8_t type = RDB_TYPE_STRING;
  fwrite(&type, sizeof(uint8_t), 1, file);
  uint32_t strlen_rdb = len;
  fwrite(&strlen_rdb, sizeof(uint32_t), 1, file);
  fwrite(str, sizeof(char), len, file);
}

// 写入键值对到 RDB 文件
void write_key_value(FILE *file, redis_map_entry *kv) {
  write_string_object(file, kv->key, strlen(kv->key));
  write_string_object(file, kv->value, strlen(kv->value));
}

// 更新 RDB 文件的 EOF 和校验和
void write_rdb_footer(FILE *file, long file_size) {
    uint64_t eof=0xFF; // EOF 标记，稍后写入
    fwrite(&eof, sizeof(uint64_t), 1, file);
    uint32_t checksum = 0; // 校验和，稍后计算并写入
    fwrite(&checksum, sizeof(uint32_t), 1, file);
}

void save_rdb_file() {
  char filepath[512];
  snprintf(filepath, sizeof(filepath), "%s/%s", config.dir, config.dbfilename);

  FILE *fp = fopen(filepath, "wb");
  if (!fp) {
    server_log(LOG_ERROR, "Failed to open RDB file for writing");
    return;
  }

  write_rdb_header(fp);

  uint8_t entrysize = 0xFB;
  fwrite(&entrysize, sizeof(uint8_t), 1, fp);
  uint8_t ucount = redis_map.size;
  fwrite(&ucount, sizeof(uint8_t), 1, fp);
  for (int i = 0; i < redis_map.size; i++) {
    uint8_t type = RDB_TYPE_STRING;
    fwrite(&type, sizeof(uint8_t), 1, fp);
    write_key_value(fp, redis_map.map[i]);
  }

  write_rdb_footer(fp, ftell(fp));

  fclose(fp);
}

// 清理资源的函数
void cleanup() {
  //dump内存rdb到文件
  save_rdb_file();
  if (epoll_fd != -1) {
    close(epoll_fd);
  }
  if (server_fd != -1) {
    close(server_fd);
  }
  if (log_fp) {
    fclose(log_fp);
  }
}

// 添加 raft 线程相关定义
static pthread_t raft_thread;
static void* raft_thread_func(void* arg) {
    int id = *(int*)arg;
    char dir[256];
    snprintf(dir, sizeof(dir), "%s", config.dir);
    init_raft(id, dir);
    return NULL;
}

int main(int argc, char *argv[]) {
  int opt;
  while ((opt = getopt(argc, argv, "d:f:i:m:")) != -1) {
    switch (opt) {
      case 'd':
        strncpy(config.dir, optarg, sizeof(config.dir) - 1);
        break;
      case 'f':
        strncpy(config.dbfilename, optarg, sizeof(config.dbfilename) - 1);
        break;
      case 'i':
        config.id = atoi(optarg);
        break;
      case 'm':
        config.m = atoi(optarg);
        break;
      default:
        fprintf(stderr, "Usage: %s [-d dir] [-f dbfilename] [-i id]\n", argv[0]);
        exit(1);
    }
  }

  if (load_rdb_file() < 0) {
    server_log(LOG_ERROR, "Failed to load RDB file");
    return 1;
  }

  // 启动 raft 线程
  pthread_create(&raft_thread, NULL, raft_thread_func, &config.id);

  // 注册信号处理函数
  struct sigaction sa;
  memset(&sa, 0, sizeof(sa));
  sa.sa_handler = handle_signal;
  sigemptyset(&sa.sa_mask);
  sa.sa_flags = SA_RESTART;  // 自动重启被中断的系统调用

  // 初始化日志
  char log_path[512];
  snprintf(log_path, sizeof(log_path), "%s/redis-server.log", config.dir);
  init_logger(log_path);

  if (sigaction(SIGTERM, &sa, NULL) == -1) {
    server_log(LOG_ERROR, "Failed to register SIGTERM handler");
    return 1;
  }
  if (sigaction(SIGINT, &sa, NULL) == -1) {
    server_log(LOG_ERROR, "Failed to register SIGINT handler");
    return 1;
  }

  // 初始化服务器
  server_fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
  if (server_fd == -1) {
    server_log(LOG_ERROR, "Socket creation failed: %s", strerror(errno));
    return 1;
  }

  // 设置 SO_REUSEADDR
  int reuse = 1;
  if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) <
      0) {
    server_log(LOG_ERROR, "SO_REUSEADDR failed: %s", strerror(errno));
    cleanup();
    return 1;
  }

  struct sockaddr_in serv_addr = {
      .sin_family = AF_INET,
      .sin_port = htons(config.m),
      .sin_addr = {htonl(INADDR_ANY)},
  };

  if (bind(server_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) != 0) {
    server_log(LOG_ERROR, "Bind failed: %s", strerror(errno));
    cleanup();
    return 1;
  }

  if (listen(server_fd, SOMAXCONN) != 0) {
    server_log(LOG_ERROR, "Listen failed: %s", strerror(errno));
    cleanup();
    return 1;
  }

  // 创建 epoll 实例
  epoll_fd = epoll_create1(0);
  if (epoll_fd == -1) {
    server_log(LOG_ERROR, "Failed to create epoll instance: %s",
               strerror(errno));
    cleanup();
    return 1;
  }

  // 添加服务器 socket 到 epoll
  struct epoll_event ev;
  ev.events = EPOLLIN;
  ev.data.fd = server_fd;
  if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &ev) == -1) {
    server_log(LOG_ERROR, "Failed to add server socket to epoll: %s",
               strerror(errno));
    cleanup();
    return 1;
  }

  server_log(LOG_INFO, "Server staring...........");

  // 主事件循环
  struct epoll_event events[MAX_EVENTS];
  while (running) {
    int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, 1000);  // 1秒超时
    if (nfds == -1) {
      if (errno == EINTR) {
        continue;  // 被信号中断，继续循环
      }
      server_log(LOG_ERROR, "epoll_wait error: %s", strerror(errno));
      break;
    }

    for (int n = 0; n < nfds && running; n++) {
      if (events[n].data.fd == server_fd) {
        // 处理新连接
        struct sockaddr_in client_addr;
        socklen_t client_addr_len = sizeof(client_addr);
        int client_fd = accept(server_fd, (struct sockaddr *)&client_addr,
                                &client_addr_len);
        if (client_fd == -1) {
          if (errno == EAGAIN || errno == EWOULDBLOCK) {
            continue;  // 没有新连接
          }
          server_log(LOG_ERROR, "Accept failed: %s", strerror(errno));
          continue;
        }
        set_nonblocking(client_fd);

        // 添加客户端 socket 到 epoll
        ev.events = EPOLLIN | EPOLLET;
        ev.data.fd = client_fd;
        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &ev) == -1) {
          server_log(LOG_ERROR, "Failed to add client to epoll: %s",
                     strerror(errno));
          close(client_fd);
          continue;
        }
        server_log(LOG_INFO, "New client connected");
      } else {
        // 处理客户端数据
        int fd = events[n].data.fd;
        char buffer[BUFFER_SIZE];
        ssize_t count = read(fd, buffer, sizeof(buffer));

        if (count == -1) {
          if (errno != EAGAIN) {
            server_log(LOG_ERROR, "Read error: %s", strerror(errno));
            epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
            close(fd);
          }
          continue;
        } else if (count == 0) {
          // 客户端断开连接
          server_log(LOG_INFO, "Client disconnected");
          epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
          close(fd);
          continue;
        }

        char command[16];
        sscanf(buffer, "*%*d\n$%*d\n%s", command);
        for (int i = 0; command[i]; i++) {
          command[i] = tolower(command[i]);
        }
        if (strncmp(command, "echo", 4) == 0) {
          char argument[256];
          sscanf(buffer, "*%*d\n$%*d\n%*s\n$%*d\n%s", argument);
          send_msg(fd, argument);
          argument[0] = '\0';
        } else if (strncmp(command, "set", 3) == 0) {
          char key[MAX_ENTRY_STR_SIZE];
          char value[MAX_ENTRY_STR_SIZE];
          // 解析 key 和 value
          sscanf(buffer,
                 "*%*d\r\n$%*d\r\nset\r\n$%*d\r\n%[^\r]%*s$%*d\r\n%[^\r]", key,
                 value);

          redis_map_entry *entry =
              (redis_map_entry *)malloc(sizeof(redis_map_entry));
          strcpy(entry->key, key);
          strcpy(entry->value, value);
          redis_map.map[redis_map.size++] = entry;

          send_msg(fd, "OK");
        } else if (strncmp(command, "get", 3) == 0) {
          char key[MAX_ENTRY_STR_SIZE];
          sscanf(buffer, "*%*d\r\n$%*d\r\nget\r\n$%*d\r\n%[^\r]", key);

          char *value = NULL;
          for (int i = 0; i < redis_map.size; i++) {
            if (strcmp(key, redis_map.map[i]->key) == 0) {
              value = redis_map.map[i]->value;
              break;
            }
          }

          if (value == NULL) {
            // Key not found, send null response
            send(fd, "$-1\r\n", 5, 0);
          } else {
            send_msg(fd, value);
          }
        } else if (strncmp(command, "config", 6) == 0) {
          char subcommand[32];
          char pattern[128];
          sscanf(buffer,
                 "*%*d\r\n$%*d\r\nconfig\r\n$%*d\r\n%[^\r]%*s$%*d\r\n%[^\r]",
                 subcommand, pattern);

          if (strcasecmp(subcommand, "get") == 0) {
            char response[4096];
            if (strcasecmp(pattern, "*") == 0) {
              // 返回所有配置
              snprintf(response, sizeof(response),
                       "*4\r\n"
                       "$3\r\ndir\r\n"
                       "$%zu\r\n%s\r\n"
                       "$9\r\ndbfilename\r\n"
                       "$%zu\r\n%s\r\n",
                       strlen(config.dir), config.dir,
                       strlen(config.dbfilename), config.dbfilename);
            } else if (strcasecmp(pattern, "dir") == 0) {
              snprintf(response, sizeof(response),
                       "*2\r\n"
                       "$3\r\ndir\r\n"
                       "$%zu\r\n%s\r\n",
                       strlen(config.dir), config.dir);
            } else if (strcasecmp(pattern, "dbfilename") == 0) {
              snprintf(response, sizeof(response),
                       "*2\r\n"
                       "$9\r\ndbfilename\r\n"
                       "$%zu\r\n%s\r\n",
                       strlen(config.dbfilename), config.dbfilename);
            } else {
              snprintf(response, sizeof(response), "$-1\r\n");  // null response
            }
            send(fd, response, strlen(response), 0);
          }
        } else {
          // 发送响应
          if (send(fd, "+PONG\r\n", 7, 0) == -1) {
            server_log(LOG_ERROR, "Send failed: %s", strerror(errno));
            epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
            close(fd);
          }
        }
      }
    }
  }

  server_log(LOG_INFO, "Server shutting down...");
  cleanup();
  // 在 cleanup() 中等待 raft 线程
  pthread_join(raft_thread, NULL);
  return 0;
}
